package com.lvlian;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.utils.OSUtils;
import com.utils.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.HashMap;
import java.util.Map;
import java.time.LocalDate;

public class LvlianSinkRecyclingIn extends RichSinkFunction<String> {
    private static final Logger LOG = LoggerFactory.getLogger(LvlianSinkRecyclingIn.class);
    private Connection connection;
    private PreparedStatement preparedStatementInsert;
    private PreparedStatement preparedStatementDelete;
    private String lastValue = null;
    private Map<Integer, KV> map = new HashMap<>();

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 加载驱动程序
        Class.forName("com.mysql.cj.jdbc.Driver");
        // 获取数据库连接
        String url = "jdbc:mysql://192.168.0.26:3306/ry-cloud";
        String user = "root";
        String password = "gni2PbgxMNutdxz";
        if (OSUtils.isWindows()) {
            url = "jdbc:mysql://110.41.6.107:33306/ry-cloud";
            user = "root";
            password = "root";
        }

        connection = DriverManager.getConnection(url, user, password);
        // 创建 PreparedStatement 对象
        String sql = "replace INTO lv_recycling_in_info ( " +
                "id,recycling_in_num,recycling_in_date,spec_name,weigh_num," +
                "reference_num,truck_num,license,contract_num,unit," +
                "spec_num,unit_price,entry_weight,rough_weight,tare_weight," +
                "deduct_weight,net_weight,amount,discount_amount,reward_amount," +
                "settlement_amount,paid_amount,pay_sts,create_user,create_date," +
                "org_code"
                + ") VALUES (" +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?" +
                ")";


        LOG.info(sql);
        preparedStatementInsert = connection.prepareStatement(sql);


        map.put(1, new KV("ROW_ID", "id"));
        map.put(2, new KV("RECYCLING_IN_NUM", "recycling_in_num"));
        map.put(3, new KV("RECYCLING_IN_DATE", "recycling_in_date"));
        map.put(4, new KV("SPEC_NAME", "spec_name"));
        map.put(5, new KV("WEIGH_NUM", "weigh_num"));
        map.put(6, new KV("REFERENCE_NUM", "reference_num"));
        map.put(7, new KV("TRUCK_NUM", "truck_num"));
        map.put(8, new KV("LICENSE", "license"));
        map.put(9, new KV("CONTRACT_NUM", "contract_num"));
        map.put(10, new KV("UNIT", "unit"));
        map.put(11, new KV("SPEC_NUM", "spec_num"));
        map.put(12, new KV("UNIT_PRICE", "unit_price"));
        map.put(13, new KV("ENTRY_WEIGHT", "entry_weight"));
        map.put(14, new KV("ROUGH_WEIGHT", "rough_weight"));
        map.put(15, new KV("TARE_WEIGHT", "tare_weight"));
        map.put(16, new KV("DEDUCT_WEIGHT", "deduct_weight"));
        map.put(17, new KV("NET_WEIGHT", "net_weight"));
        map.put(18, new KV("AMOUNT", "amount"));
        map.put(19, new KV("DISCOUNT_AMOUNT", "discount_amount"));
        map.put(20, new KV("REWARD_AMOUNT", "reward_amount"));
        map.put(21, new KV("SETTLEMENT_AMOUNT", "settlement_amount"));
        map.put(22, new KV("PAID_AMOUNT", "paid_amount"));
        map.put(23, new KV("PAY_STS", "pay_sts"));
        map.put(24, new KV("RECORDER", "create_user"));
        map.put(25, new KV("RECORD_DATE", "create_date"));
        map.put(26, new KV("ORG_CODE", "org_code"));


    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 将 JSON 字符串解析为 JSONObject
        JSONObject jsonObject = JSON.parseObject(value);

        // 获取操作类型
        String operation = jsonObject.getString("op");

        // 判断操作类型
        if ("d".equals(operation)) {
            // 删除操作
            JSONObject beforeObject = jsonObject.getJSONObject("before");
            String id = beforeObject.getString("ROW_ID");
            preparedStatementDelete.setString(1, id);
            preparedStatementDelete.executeUpdate();
        } else {
            // 插入或更新操作
            JSONObject afterObject = jsonObject.getJSONObject("after");
            // 如果上一次处理的值为空，或者当前值与上一次处理的值不相同，则执行 sink 操作
            if (lastValue == null || !lastValue.equals(value)) {
                // 遍历原始表字段
                for (Map.Entry<Integer, KV> entry : map.entrySet()) {
                    String v = afterObject.getString(entry.getValue().getK());


//                if ("PAY_TIME".equals(entry.getValue().getK()) || "RECORD_DATE".equals(entry.getValue().getK())) {
                    if ("RECYCLING_IN_DATE".equals(entry.getValue().getK()) || "RECORD_DATE".equals(entry.getValue().getK())) {
                        v = StringUtils.restoreDate(v);
                    }
                    LOG.info(entry.getKey() + ": " + entry.getValue().getK() + " -> " + entry.getValue().getV() + " -> " + v);
                    preparedStatementInsert.setString(entry.getKey(), v);
                }

                preparedStatementInsert.executeUpdate();

                // 更新上一次处理的值为当前值
                lastValue = value;
            }
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭连接和 PreparedStatement
        if (preparedStatementInsert != null) {
            preparedStatementInsert.close();
        }
        if (preparedStatementDelete != null) {
            preparedStatementDelete.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
